package org.tiger.springboot.rocketMq.consumer;

import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * 消费者
 * 
 * @author liuwenlai
 *
 */
@Component
public class RocketMqConsumer implements MessageListenerConcurrently {

	private static Logger logger = LoggerFactory.getLogger(RocketMqConsumer.class);

	@Value("${spring.rocketmq.namesrvaddr}")
	private String namesrvAddr;

	private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TestRocketMQPushConsumer");

	/**
	 * 订阅主题
	 */
	@Value("${spring.rocketmq.subscribe}")
	private String subscribe;

	@Value("${spring.rocketmq.messageSelector}")
	private String messageSelector;

	/**
	 * 初始化消费者
	 */
	@PostConstruct
	public void start() {
		try {
			logger.info("########### MQ consumer start");

			consumer.setNamesrvAddr(namesrvAddr);
			// 消息队列从头开始消费
			consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
			// 消费模式 集群
			consumer.setMessageModel(MessageModel.CLUSTERING);
			// subscribe : topic
			// messageSelector : tag
			consumer.subscribe(subscribe, messageSelector);
			// 注册消息监听器
			consumer.registerMessageListener(this);
			// 启动
			consumer.start();
		} catch (MQClientException e) {
			logger.error("########## MQ consumer start error", e);
		}
	}

	/**
	 * 
	 */
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> message, ConsumeConcurrentlyContext context) {
		logger.info("message size ########################:" + message.size());
		for (int i = 0; i < message.size(); i++) {
			MessageExt messageExt = message.get(i);
			// TODO ：inner or outer try
			try {
				String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
				logger.info("MQ：消费者接收新信息: {} {} {} {} {}", messageExt.getMsgId(), messageExt.getTopic(),
						messageExt.getTags(), messageExt.getKeys(), messageBody);
			} catch (Exception e) {
				logger.error("consumer get message error:", e);
			}
		}
		context.setAckIndex(message.size() + 1);

		// 另一种写法 区别：context.setAckIndex(index + 1);
		// int index = 0;
		// try {
		// for (; index < msgs.size(); index++) {
		// MessageExt msg = msgs.get(index);
		//
		// String messageBody = new String(msg.getBody(),
		// RemotingHelper.DEFAULT_CHARSET);
		//
		// LOGGER.info("MQ：消费者接收新信息: {} {} {} {} {}", msg.getMsgId(),
		// msg.getTopic(), msg.getTags(), msg.getKeys(), messageBody);
		// }
		// } catch (Exception e) {
		// LOGGER.error(e.getMessage(), e);
		// } finally {
		// if (index < msgs.size()) {
		// context.setAckIndex(index + 1);
		// }
		// }

		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}

	/**
	 * 关闭消费者
	 */
	@PreDestroy
	public void stop() {
		if (consumer != null) {
			consumer.shutdown();
			logger.info("########### MQ consumer shutdown ");
		}
	}

}
